package com.dark;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Created by tengxue on 16-7-19.
 * 简单的生产者和消费者的示例
 */
public class SimpleDemo {

    private static final Logger LOG = LoggerFactory.getLogger(SimpleDemo.class);

    public static void main(String[] args){
        String brokers = "localhost:9092";
        String zookeepers = "localhost:2181";
        String topic = "test-replication";

        // 启动生产者
        startProducer(brokers, topic);

        // 启动消费者
        startConsumer(zookeepers, topic);
    }

    /** 启动生产者 */
    private static void startProducer(final String brokers, final String topic) {
        new Thread() {
            public void run() {
                final Properties props = new Properties();
                props.put("serializer.class", "kafka.serializer.StringEncoder");
                props.put("metadata.broker.list", brokers);
                // 生产者
                kafka.javaapi.producer.Producer<String, String> producer = new kafka.javaapi.producer.Producer<String, String>(new ProducerConfig(props));

                for (int i = 0; i < 10; i++) {
                    for (int j = 0; j < 100; j++) {
                        String key = i + "_" + j;
                        String message = "Message_" + key;
                        // 发送消息
                        producer.send(new KeyedMessage<String, String>(topic, key, message));
                        LOG.debug("send, topic: {}, message: {}", topic, message);
                    }
                }
            }
        }.start();
    }

    /** 启动消费者 */
    private static void startConsumer(final String zookeepers, final String topic) {
        new Thread() {
            public void run() {
                final Properties props = new Properties();
                props.put("zookeeper.connect", zookeepers);
                // 消费者
                ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
                Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
                topicCountMap.put(topic, new Integer(1));
                Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
                // topic消息流
                KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
                ConsumerIterator<byte[], byte[]> it = stream.iterator();
                while (it.hasNext()) {
                    MessageAndMetadata<byte[], byte[]> msgData = it.next();
                    String message = new String(msgData.message());
                    LOG.debug("received, message:{}", message);
                    // do something
                }
            }
        }.start();
    }
}
